package top.hypnos.bigdata.filesystem;

import top.hypnos.bigdata.filesystem.command.CommandImpl;
import top.hypnos.bigdata.filesystem.record.ObjectInputStreamImpl;
import top.hypnos.bigdata.filesystem.record.ObjectOutputStreamImpl;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

public class FileBinLog implements BinLog {

    private final File basePath;
    private int fileIndex = 0;
    private FileOutputStream output;

    public FileBinLog(final File basePath) throws IOException {
        this.basePath = basePath;
        if (!basePath.exists()) {
            if (!basePath.mkdirs()) {
                throw new IOException("无法创建目录：" + basePath.getAbsolutePath());
            }
        }
    }

    @Override
    public int append(final Command command) throws IOException {
        final var txnId = command.getTxnId();
        final var fileIndex = (txnId + 999) / 1000;
        if (this.fileIndex != fileIndex) { // 切换文件输出流
            if (output != null) {
                try {
                    output.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            final var file = new File(basePath, "binlog-" + String.format("%03d", fileIndex) + ".bin");
            output = new FileOutputStream(file);
            this.fileIndex = fileIndex;
        }
        command.serialize(new ObjectOutputStreamImpl(output));
        return txnId;
    }

    @Override
    public Collection<Command> loadAfter(int txnId) throws IOException {
        // TODO 异常结束导致binlog最后一条日志不完整问题，待处理
        final List<Command> result = new ArrayList<>();
        final var fileIndexFrom = txnId / 1000 + 1;
        for (var fileIndex = fileIndexFrom; ; fileIndex++) {
            final var file = new File(basePath, "binlog-" + String.format("%03d", fileIndex) + ".bin");
            if (!file.exists()) {
                break;
            }
            try (final var input = new FileInputStream(file)) {
                while (input.available() > 0) {
                    try {
                        final var command = new CommandImpl<>();
                        command.deserialize(new ObjectInputStreamImpl(input));
                        if (command.getTxnId() > txnId) {
                            result.add(command);
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
        return result;
    }
}
